// Released in the Public Domain. http://creativecommons.org/licenses/publicdomain/
package org.xi8ix.async.spi;

import org.xi8ix.async.AsyncEngine;
import org.xi8ix.async.Receiver;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.*;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Future;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * @author Iain Shigeoka
 */
class SocketPeer extends AbstractPeer implements SendFutureSender {
    /** Class-wide logger. */
    private static final Logger LOG = Logger.getLogger(SocketPeer.class.getName());
    // The overhead in bytes of the IP header
    private static final int IP_HEADER_SIZE = 20;

    private boolean closed;
    private AsyncEngine engine;
    private SocketChannel channel;
    private InetSocketAddress localAddress;
    private Receiver receiver;
    private Queue<SendFuture<AbstractPeer>> roundRobinQueue = new ConcurrentLinkedQueue<SendFuture<AbstractPeer>>();

    public SocketPeer(InetSocketAddress localAddress, InetSocketAddress remoteAddress, Receiver receiver, AsyncEngine engine) throws IOException {
        super(remoteAddress);

        this.localAddress = localAddress;
        this.receiver = receiver;
        this.engine = engine;

        channel = SocketChannel.open();
        if (localAddress != null) {
            channel.socket().bind(localAddress);
        }
        channel.connect(remoteAddress);
        setup();
    }

    public SocketPeer(SocketChannel channel, Receiver receiver, AsyncEngine engine) throws IOException {
        super((InetSocketAddress) channel.socket().getRemoteSocketAddress());
        this.localAddress = (InetSocketAddress) channel.socket().getLocalSocketAddress();
        this.channel = channel;
        this.receiver = receiver;
        this.engine = engine;
        setup();
    }

    private void setup() throws ClosedChannelException {
        engine.add(channel, SelectionKey.OP_READ, selectableReadCallback);
        tuner = new TunerImpl(1024){
            public int getRecvBufferSize() {
                try {
                    return channel.socket().getReceiveBufferSize();
                } catch (SocketException e) {
                    LOG.log(Level.WARNING, "Unable to get buffer size", e);
                    return -1;
                }
            }

            public void setRecvBufferSize(int recvBufferSize) {
                try {
                    channel.socket().setReceiveBufferSize(recvBufferSize);
                } catch (SocketException e) {
                    LOG.log(Level.WARNING, "Unable to get buffer size", e);
                }
            }
        };
    }

    /**
     * Closes the socket associated with this object, removes its callbacks
     * from ASyncCore, and causes it to stop responding to any outstanding
     * timers it has registered--in other words, turns it off.  There is no
     * way to turn it back on; instead, just create a new one.
     *
     * @throws IOException if there's a problem closing this sender
     */
    public void close() throws IOException {
        engine.remove(channel);
        channel.socket().close(); // closes channel with it
        closed = true;
    }

    /**
     * Send data to another host.
     *
     * @param dst the message's destination address
     * @param data the data to send
     * @return a token than is used to cancel the send and receive it's successful completion status.
     */
    public Future<Boolean> send(InetSocketAddress dst, ByteBuffer data) {

        LOG.log(Level.FINEST,"sending {0} -> {1}",new Object[]{data,dst});

        SendFuture<AbstractPeer> future = new SendFuture<AbstractPeer>(this, this, data);
        getSendQueue().add(future);
        LOG.log(Level.FINEST,"peer for {0}: Q={1}", new Object[]{dst, getSendQueue()});

        if (!getSendQueue().isEmpty()) {
            roundRobinQueue.add(future);
            try {
                engine.add(channel,SelectionKey.OP_READ | SelectionKey.OP_WRITE, selectableWriteCallback);
            } catch (ClosedChannelException e) {
                LOG.log(Level.WARNING,"Trouble adding write op",e);
            }
        } else {
            LOG.finest("not yet isSendable");
        }

        return future;
    }

    /**
     * Handles reading from the channel when data is available.
     *
     * @author Iain Shigeoka
     */
    private Runnable selectableReadCallback = new Runnable() {
        /**
         * Reads available data from the channel. The operation only needs to read one
         * packet since the AsyncEngine will re-run this this method if more data is available.
         */
        public void run() {
            if (!closed) {

                ByteBuffer buffer = ByteBuffer.allocate(tuner.getBufferSize());
                try {
                    int read = channel.read(buffer);
                    if (read > 0) {
                        buffer.flip();

                        // Update port statistics
                        long now = System.currentTimeMillis();
                        int inboundByteTotal = buffer.remaining() + IP_HEADER_SIZE;
                        statistics.addInboundBytes(inboundByteTotal, 1);
                        statistics.setLastRecv(now);

                        // Send to receiver
                        receiver.onRecv(buffer, address, localAddress);
                    }
                } catch (IOException e) {
                    LOG.log(Level.FINEST, "", e);
                }
            }
        }
    };

    /**
     * Handles writing to the channel when data is queued for sending.
     */
    private Runnable selectableWriteCallback = new Runnable() {
        /**
         * Writes data to the channel. The operation only needs to write one packet
         * since the AsyncEngine will re-run this method if the channel is write-ready.
         * When the queue is empty, we'll cancel our interest in the write operation to
         * prevent unnecessary calls to this method.
         */
        public void run() {
            // Poll is a non-blocking remove of the head of the queue
            SendFuture<AbstractPeer> future = roundRobinQueue.poll();
            if (future == null) {
                // No pending futures waiting to be sent... let's remove the operation
                try {
                    engine.add(channel,SelectionKey.OP_READ,selectableReadCallback);
                } catch (ClosedChannelException e) {
                    LOG.log(Level.SEVERE,"Problem re-registering channel",e);
                }
            } else {
                // A valid packet ready to be sent, let's send it
                long now = System.currentTimeMillis();
                AbstractPeer peer = future.getPeer();

                ByteBuffer data = future.getData();
                int sz = data.remaining();
                if (sz > tuner.getBufferSize()) {
                    LOG.finest("size=" + sz + " is greater than max size=" + tuner.getBufferSize() + " for msg " + data);
                }

                int outboundByteTotal = data.remaining() + IP_HEADER_SIZE;

                try {
                    channel.write(data);
                    if (data.remaining() == 0) {
                        // Send was successful.
                        peer.getSendQueue().remove();

                        statistics.addOutboundBytes(outboundByteTotal, 1);
                        statistics.addTimeToFirstSend(now - future.getStartTime());
                        statistics.setLastSend(now);

                        peer.getStatistics().addOutboundBytes(outboundByteTotal, 1);
                        peer.getStatistics().addTimeToFirstSend(now - future.getStartTime());
                        peer.getStatistics().setLastSend(now);
                    }
                    // what should we do if not all the data was sent?
                    // Our protocols don't support data split across multiple packets so we simply call it quits
                } catch (IOException e) {
                    LOG.log(Level.SEVERE,"Send failed",e);
                }
            }
        }
    };

    /**
     * Adds the UUID to the list of messages to cancel.
     *
     * @param id the ID to cancel
     */
    public void cancel(UUID id) {
        Iterator<SendFuture<AbstractPeer>> iterator = roundRobinQueue.iterator();
        while (iterator.hasNext()) {
            if (iterator.next().getId().equals(id)) {
                iterator.remove();
            }
        }
    }
}
